/*************************************************************************************
* 	 Copyright (C) 2010 by Information Systems Group, Saarland University  			*
*    http://infosys.cs.uni-saarland.de												*
* 	 																				*
* 	 This file is part of Hadoop++.												 	*
*																					*
*    Hadoop++ is free software: you can redistribute it and/or modify				*
*    it under the terms of the GNU Lesser General Public License as published by	*
*    the Free Software Foundation, either version 3 of the License, or				*
*    (at your option) any later version.											*
*																					*
*    Hadoop++ is distributed in the hope that it will be useful,					*
*    but WITHOUT ANY WARRANTY; without even the implied warranty of					*
*    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the					*
*    GNU Lesser General Public License for more details.							*
*																					*
*    You should have received a copy of the GNU Lesser General Public License		*
*    along with Hadoop++.  If not, see <http://www.gnu.org/licenses/>.				*
*************************************************************************************/
package unisb.cs.core.index.access;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.io.IOUtils;

import unisb.cs.core.binary.utils.BinaryUtils;
import unisb.cs.core.index.structure.CSSTree;
import unisb.cs.core.index.structure.IndexHeader;
import unisb.cs.data.tables.Record;
import unisb.cs.data.tables.TableObjectProxy;
import unisb.cs.data.types.DataTypes;

/**
 * This class looks up the index and creates a record iterator
 * to access relevant portion of data within the split.
 *
 */
public class IndexLookup {
	// index variables
	int key, keyType, keySize;
	String	tableName;
	CSSTree cssTree = null;

	int tupleOffset = -1;
	int contentSize = 0;

	public IndexLookup(int key, int keyType, int keySize, String tableName) {
		this.key = key;
		this.keyType = keyType;
		this.keySize = keySize;
		this.tableName = tableName;
	}

	/**
	 * This method extracts the relevant data from the input split and 
	 * returns an iterator over the relevant records.
	 * 
	 * @param in The split input data stream
	 * @param lowKey The low key for index access
	 * @param highKey The high key for index access
	 * @param dataOffset The data offset to start index access from
	 * @param splitEndPosition The last offset in this split
	 * @return An iterator over the relevant records within this split. 
	 * @throws IOException
	 */
	public RecordList extract(FSDataInputStream in, String lowKey, String highKey, long dataOffset, long splitEndPosition) throws IOException {
		// Read the index header
		IndexHeader idxHeader = new IndexHeader(keyType, keySize);
		byte[] header = new byte[idxHeader.getHeaderSize()];
		in.seek(splitEndPosition - idxHeader.getHeaderSize() + 1);
		IOUtils.readFully(in, header, 0, idxHeader.getHeaderSize());
		idxHeader.parse(header);
				
		// Read index (only if required)
		in.seek(splitEndPosition - idxHeader.getHeaderSize() - idxHeader.getIndexSize() + 1);
		readIndex(in, lowKey, highKey, idxHeader);

		if (tupleOffset >= 0) {
			// relevant data
			in.seek(dataOffset + tupleOffset);
			return new RecordList(in, tupleOffset, contentSize, lowKey, highKey, tableName);
		} else
			return null;	// no relevant data found in the split
	}

	/**
	 * Read the index only if the split contains the relevant data.
	 * This method compares the query low/high keys with the low/high 
	 * keys in the index header. The method does an actual index lookup 
	 * only if the spit contains relevant data.
	 *  
	 * @param in The split data input stream
	 * @param lowKey The query low key
	 * @param highKey The query high key
	 * @param idxHeader The index header
	 * @throws IOException
	 */
	//ʹ��indexʱ�����ж�query��range��index��range�Ĺ�ϵ��Ȼ���ȡ��ʼ��¼��λ�ü��ܹ�������
	//����Ϳ��Դ���ʼ��¼��ʼ��ȡ�ܹ�������ô��ļ�¼����
	private void readIndex(FSDataInputStream in, String lowKey, String highKey, IndexHeader idxHeader) throws IOException {
		// Read Index
		int overlapType = idxHeader.getOverlapType(lowKey, highKey);
		System.out.println(lowKey + highKey + "Overlap Type:"+overlapType);
		if (overlapType == IndexHeader.SPANS || overlapType == IndexHeader.RIGHT_CONTAINED) {
			// relevant data
			byte[] index = new byte[idxHeader.getIndexSize()];
			IOUtils.readFully(in, index, 0, idxHeader.getIndexSize());
			cssTree = new CSSTree(index, idxHeader.getNofTuples(), 5, keySize, keyType, keySize, 0);
			tupleOffset = 0;
			contentSize = idxHeader.getIntDataSize();
		} else if (overlapType != IndexHeader.NOT_CONTAINED) {
			// relevant data
			byte[] index = new byte[idxHeader.getIndexSize()];
			IOUtils.readFully(in, index, 0, idxHeader.getIndexSize());
			cssTree = new CSSTree(index, idxHeader.getNofTuples(), 5, keySize, keyType, keySize, 0);
			tupleOffset = cssTree.get(lowKey);
			contentSize = idxHeader.getIntDataSize();
		} else {
			// no relevant data in the split
			tupleOffset = -1;
		}
	}

	/**
	 * This class iterates over the relevant records within the split.
	 * i.e. after locating the starting offset of the low key, we do a 
	 * sequential scan in the split till we reach the high key in the query. 
	 *
	 */
	public class RecordList implements Iterator<Record> {
		long offset, splitSize;
		FSDataInputStream in;
		byte[] highKeyBytes;

		private byte[] data;
		
		/** variable to prefetch and store the next record */
		private Record nextRecord = null;
		private boolean fetched = false;
		private String tableName;
		/**
		 * Create a record list.
		 * @param in The split data input stream
		 * @param offset The starting offset within the split
		 * @param splitSize The split size
		 * @param lowKey The low key in the query
		 * @param highKey The high key in the query
		 */
		public RecordList(FSDataInputStream in, long offset, long splitSize, String lowKey, String highKey, String tableName) {
			this.offset = offset;
			this.splitSize = splitSize;
			this.in = in;
			this.tableName = tableName;

			highKeyBytes = getKeyBytes(highKey);
			nextRecord = fetchNext();
			byte[] lowKeyBytes = getKeyBytes(lowKey);
			while (nextRecord != null && cssTree.compareKeys(nextRecord.getByteAttribute(key), lowKeyBytes) < 0)
				nextRecord = fetchNext();
			fetched = true;
		}

		/**
		 * Convert key into byte array
		 */
		private byte[] getKeyBytes(String key) {
			switch (keyType) {
			case DataTypes.INT_TYPE:
				return BinaryUtils.toBytes(Integer.parseInt(key));
			case DataTypes.FLOAT_TYPE:
				return BinaryUtils.toBytes(Float.parseFloat(key));
			case DataTypes.CHAR_TYPE:
				return key.getBytes();
			case DataTypes.DATE_TYPE:
				return BinaryUtils.toBytes(key);
			default:
				return null;
			}
		}

		/**
		 * Do we have more relevant records within the split. 
		 */
		public boolean hasNext() {
			if (!fetched) {
				nextRecord = fetchNext();
				fetched = true;
			}
			return nextRecord == null ? false : true;
		}

		/**
		 * Fetch the next relevant record in the split.
		 */
		public Record next() {
			if (!fetched)
				nextRecord = fetchNext();
			else
				fetched = false;
			return nextRecord;
		}

		/**
		 * Skip the remaining data within the split.
		 * This method is useful when the split contains
		 * co-grouped data and we want to do index access 
		 * over one of the relations 
		 */
		public void skipRest() {
			try {
				if ((splitSize - offset) > 0)
					IOUtils.skipFully(in, splitSize - offset);
			} catch (IOException e) {
				e.printStackTrace();
			}
		}

		/**
		 * skip one record
		 */
		public void remove() {
			fetchNext();
		}

		/**
		 * Read the next record from the input stream
		 * @return The next record
		 */
		private Record fetchNext() {
			if (offset < splitSize) {
				try {
					TableObjectProxy proxy = new TableObjectProxy(tableName);
					int recordLen = proxy.getSize();
					if (recordLen == 0)
						return null;
					if (offset > (splitSize - recordLen))
						return null;

					data = new byte[recordLen];
					IOUtils.readFully(in, data, 0, recordLen);
					offset += recordLen;
					proxy.setRecord(data);
					Record r = proxy.getRecord();
					return (cssTree.compareKeys(proxy.getByteAttribute(key), highKeyBytes) > 0) ? null : r;
				} catch (Exception e) {
					e.printStackTrace();
				}

			}
			return null;
		}
	}
}
